package cn.linjianhui.flink.sample.function;

import cn.linjianhui.flink.sample.model.OmsOrder;
import cn.linjianhui.flink.sample.model.OrderAccumulator;
import org.apache.flink.api.common.functions.AggregateFunction;

import java.math.BigDecimal;

public class OrderAggregateFunction implements AggregateFunction<OmsOrder, OrderAccumulator, OrderAccumulator> {
    @Override
    public OrderAccumulator createAccumulator() {
        return new OrderAccumulator();
    }

    @Override
    public OrderAccumulator add(OmsOrder order, OrderAccumulator accumulator) {
        if("".equals(accumulator.getOrderSource())) {
            accumulator.setOrderSource(String.valueOf(order.getORDERSOURCE()));
        }

        if (order.getORDERSTATUS() == 3) {
            if(!accumulator.containsOrder(order.getORDERCODE())) {
                accumulator.addPayOrder(1L);
                accumulator.addPayAmount(new BigDecimal(order.getNETAMT()));
                accumulator.addOrderId(order.getORDERCODE());
            }
        } else if (order.getORDERSTATUS() == 10) {
            if(!accumulator.containsCancelOrder(order.getORDERCODE())) {
                accumulator.addCancelOrderNum(1L);
                accumulator.addCancelPayAmount(new BigDecimal(order.getNETAMT()));
                accumulator.addCancelOrderId(order.getORDERCODE());
            }
        }

        return accumulator;
    }

    @Override
    public OrderAccumulator getResult(OrderAccumulator result) {
        return result;
    }

    @Override
    public OrderAccumulator merge(OrderAccumulator a, OrderAccumulator b) {
        if("".equals(a.getOrderSource())) {
            a.setOrderSource(b.getOrderSource());
        }
        a.addPayAmount(b.getTotalPayAmount());
        a.addPayOrder(b.getTotalPayOrder());
        a.addOrderIds(b.getOrderIds());
        a.addCancelOrderNum(b.getCancelOrderNum());
        a.addCancelPayAmount(b.getCancelPayAmount());
        a.addCancelOrderIds(b.getCancelOrderIds());
        return a;
    }
}
